package cc.verywell.pureblock.func.node;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import cc.verywell.pureblock.func.node.NodeAlias.Alias;
import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.net.AioClientConnector;
import cc.verywell.pureblock.net.NormalSession;

/**
* <pre> 
* +++++++++++++++
* +    Entry    +
* +++++++++++++++
* + Key + Value +
* +++++++++++++++
* </pre>
* <p>
* 
* 
* ok, memory low.
* 50万邻居排序7秒返回，100万邻居内存会溢出(limit500Mb);
* 如有幸发布至公链使用，请务必重载Top-K算法;
* 
* 
* </p>
* Copyright © 2019 TanYaqiu. All rights reserved.
* @author TanYaqiu 
* @address flat 601
* @date 2019年3月23日 下午4:28:36
*/
public class NodeBucket {
	static public int NEW_KEYS_THROWED = 16;
	static public int GC_START = 128*1024; //每大于13万邻居就要进行一次清理, 有生之年不会遇到的哈;

//	static private NodeBucket instance = new NodeBucket();
//	static public NodeBucket getInstance() {
//		return instance;
//	}
	
	final protected Map<KeyShell,ValueShell> map;
	final private ValueShell myNode;
	public NodeBucket(Node myNode) {
		this(new ValueShell(new NodeStock(myNode)));
	}
	public NodeBucket(ValueShell myNode) { ///对func内部使用;
		map = new ConcurrentHashMap<>();
		newKey = new LinkedBlockingQueue<>(NEW_KEYS_THROWED);
		newKeyListener = new HashSet<>(0);
		{
			ValueShell v = myNode;
			if(null==map.putIfAbsent(v.key, v)) {
				this.myNode = v;
			}else {
				throw new IllegalArgumentException();
			}
		}
	}
	public ValueShell getMyNode() {
		return myNode;
	}
	public void save() {
		//TODO
		Collection<ValueShell> sort = map.values();//getTargetNearby(myNode.getPubKey(),map.size());
		if(Pblog.DEBUG) {
			for (ValueShell v : sort) {
				Pblog.DEBUG(distance(myNode.getPubKey(), v.getPubKey()),"-",v.node.node,",lastConnect:",v.node.lastConnect,",lastRemote:",v.node.lastRemote);
			}
		}
	}
	
	/**
	 * 
	 * @param msg
	 * @param session
	 * @param time
	 * @return null if node was connected, ValueShell is ok;
	 */
	public synchronized ValueShell allowNodeConnect(Node msg ,NormalSession session, long time) {
		NodeStock node = new NodeStock(msg, session, time);
		ValueShell v = new ValueShell(node);
		ValueShell org = map.putIfAbsent(v.key, v);
		if(org==null) {
			while(!newKey.offer(v)) {
				throwNewKey();
			}
			throwNewKey();
			return v;
		}else if(org==myNode) {
			return null;
		}
		synchronized (org) {
			if(org.key.key==null) { //此org已被其它线程删除;
				return allowNodeConnect(msg, session, time);
			}
			NormalSession ns = org.node.getSession();
			if(ns!=null && !ns.isClosed()) {
				return org;
			}
			org.node = node;
			org.key.key = node.node.pubKey;
			return org;
		}
	}

	public void addNode(Node... node) {
		if(node.length+map.size()>GC_START) { //清理
			synchronized (map) {
				int much = node.length+map.size()-GC_START;
				if(much>0) {
					ArrayList<ValueShell> sort = new ArrayList<ValueShell>(map.values());
					Collections.sort(sort,new Comparator<ValueShell>() {
						public int compare(ValueShell o1, ValueShell o2) {
							long t = Math.max(o1.node.node.getRemoteSentTime(),o1.node.lastConnect) - Math.max(o2.node.node.getRemoteSentTime(),o2.node.lastConnect);
							if(t!=0)return t>0?1:-1;
							return 0;
						}
					} );
					much = Math.max(much*2, map.size()/5);
					for (ValueShell v : sort) {
						synchronized (v) {
							NormalSession s = v.node.getSession();
							if(s!=null && !s.isClosed())continue;
							if(v==myNode)continue;
							map.remove(v.key);
							v.key.key=null; //通知其它,这个ValueShell已经停止使用了.
							if(--much==0)break;
						}
					}
				}
			}
			
		}
		for (Node n : node) {
			ValueShell v = new ValueShell(new NodeStock(n));
			ValueShell org = map.putIfAbsent(v.key, v);
			if(org==null) {
				while(!newKey.offer(v)) {
					throwNewKey();
				}
				continue;
			}
			synchronized (org) {
				if(org.key.key==null) { //此org已被其它线程删除;
					continue;
				}
				NormalSession ns = org.node.getSession();
				if(ns!=null && !ns.isClosed())continue;
				if(n.getRemoteSentTime()>org.node.node.getRemoteSentTime()) {
					org.node.node = n;
					org.key.key = n.getPubKey();
					map.putIfAbsent(org.key, org); //重新再put?
				}
			}
		}
		if(!newKey.isEmpty())throwNewKey();
	}
	
	/**
	 * 
	 */
	public interface NewKeyListener{
		public void newKey(Collection<ValueShell> values);
	}
	public void addNewKeyListener(NewKeyListener l) {
		synchronized (newKeyListener) {
			newKeyListener.add(l);
		}
	}
	public void removeNewKeyListener(NewKeyListener l) {
		synchronized (newKeyListener) {
			newKeyListener.remove(l);
		}
	}
	private Collection<NewKeyListener> newKeyListener;
	private BlockingQueue<ValueShell> newKey;
	private void throwNewKey() {
		Collection<ValueShell> al = new ArrayList<>(NEW_KEYS_THROWED); 
		if(0==newKey.drainTo(al))return;
		synchronized (newKeyListener) {
			for (NewKeyListener l : newKeyListener) {
				l.newKey(al);
			}
		}
	}
	
	/**
	 * <p>prior return connecting or had connected node</p>;
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 * @param KEY
	 * @param MAXreturn
	 * @return
	 */
	public List<ValueShell> getTargetNearby(final byte[] KEY,final int MAXreturn){
		if(MAXreturn<=0)return Collections.emptyList();//new ArrayList<>(0);
		Comparator<ValueShell> c = new Comparator<ValueShell>() {
			public int compare(ValueShell o1, ValueShell o2) {
//在func里面另外再搞一个bucket;
//				WeakReference<NormalSession> w1,w2;
//				w1 = o1.node.weakSession;
//				w2 = o2.node.weakSession;
//				if(w1!=null) {
//					if(w2==null)return -1;
//					NormalSession n1,n2;
//					n1 = w1.get();
//					n2 = w2.get();
//					if(n1!=null && !n1.isClosed()) {
//						if(n2==null || n2.isClosed())return -1;
//						//其他情况照旧查distance;
//					}else if(n2!=null && !n2.isClosed()) {
//						return 1;
//					}
//					//最近链结时间;
					long t;
					//t = o1.node.lastConnect>>>26 - o2.node.lastConnect>>>26; //?? approx 0.7-1.6 day;
					//t = o1.node.lastConnect>>>25 - o2.node.lastConnect>>>25; //?? approx 4-9 hour;
					t = o1.node.node.getRemoteSentTime()>>>29 - o2.node.node.getRemoteSentTime()>>>29; //?? approx 4-6 day;
					if(t!=0)return t>0?1:-1;
//					//其他情况照旧查distance;
//				}else if(w2!=null)return 1;
				//return distance(key,o1.key.key)-distance(key,o2.key.key);
				boolean o1connectable = ValueShell.isConnectable(o1.getRemote()); 
				if(o1connectable!=ValueShell.isConnectable(o2.getRemote()))return o1connectable?-1:1;
				return o1.getDistance(KEY)-o2.getDistance(KEY);
			}
		};
		
		Collection<ValueShell> vs  = map.values();
		ValueShell[] ds = new ValueShell[Math.min(vs.size(), MAXreturn*3)];
		int i=0, topk = Math.min(MAXreturn, ds.length);
		ValueShell p = null;
		for (ValueShell v : vs) {
			if(v==myNode)continue; //自身除外;
			if(p==null || 0>c.compare(p, v)) {
				ds[i++]=v;
				if(i==ds.length) {
					Arrays.sort(ds, c);
					i = topk;
					p = ds[i-1];
				}
			}
		}
		if(i!=topk)Arrays.sort(ds,0,i,c);
		
		return Arrays.asList(ds).subList(0, Math.min(i, topk));
	}
	
	/**
	 * 欢迎重载 Top-K 算法;
	 * 
	 * <p>性能对比，见test目录下{@link NodeBucketTest.java}</p>
	 * <p>有更好方案微信(gzpf2288)我一起交流吧。</p>
	 * 
	 * Copyright © 2019 TanYaqiu. All rights reserved.
	 */
	public Node[] getNeighbourNodes(final byte[] KEY,final int MAXreturn){
		if(MAXreturn<=0)return new Node[0];
		Comparator<ValueShell> c = new Comparator<ValueShell>() {
			public int compare(ValueShell o1, ValueShell o2) {
				int i = o1.getDistance(KEY)-o2.getDistance(KEY);
				if(i!=0)return i;
				long t = o1.node.node.getRemoteSentTime()-o2.node.node.getRemoteSentTime();
				return t>0?-1:t==0?0:1; //基本有序
			}
		};
		Collection<ValueShell> vs  = map.values();
		ValueShell[] ds = new ValueShell[Math.min(vs.size(), MAXreturn*3)];
		int i=0, p=Integer.MAX_VALUE, topk = Math.min(MAXreturn, ds.length);
		long remoteSent=0;
		for (ValueShell v : vs) {
			if(!ValueShell.isConnectable(v.getRemote()) || v==myNode)continue; //要有对外接口 and 不是自己
			int s = v.getDistance(KEY);
			if(s<=p) {
				if(s==0)continue; //不是对方
				if(s==p && remoteSent>=v.node.node.getRemoteSentTime())continue;
				ds[i++]=v;
				if(i==ds.length) {
					Arrays.sort(ds, c);
					i = topk;
					v = ds[i-1];
					p = v.getDistance(KEY);
					remoteSent = v.node.node.getRemoteSentTime();
				}
			}
		}
		if(i!=topk)Arrays.sort(ds,0,i,c);
		
		Node[] result = new Node[Math.min(i, topk)];
		i=0;
		for (ValueShell v : ds) {
			if(i==result.length)break;
			result[i++]=v.node.node;
		}
		return result;
	}
	
	//////Simple distance
	final static public int distance(final byte[] myKey, final byte[] yourKey) {
		int distance = 0;
		int min = Math.min(myKey.length, yourKey.length);
		for (int i = 0; i < min; i++) {
			int b = (0xFF & myKey[i]) ^ (0xFF & yourKey[i]);
			if((b&1)!=0)distance++;
			if((b&2)!=0)distance++;
			if((b&4)!=0)distance++;
			if((b&8)!=0)distance++;
			if((b&16)!=0)distance++;
			if((b&32)!=0)distance++;
			if((b&64)!=0)distance++;
			if((b&128)!=0)distance++;
		}
		if(min!=myKey.length)  distance += (myKey.length-min)*8;
		if(min!=yourKey.length)distance += (yourKey.length-min)*8;
		return distance;
	}
	
	///////////Sub class
	public static class ValueShell implements AioClientConnector{
		final public KeyShell key;
		NodeStock node;
		private Alias alias;
		public ValueShell(NodeStock n) {
			this.node = n;
			this.key = new KeyShell(n.node.getPubKey());
			this.alias = NodeAlias.map.get(this.key);
		}
		private transient DistanceBuffer distanceBuffer;
		final public int getDistance(byte[] b) {
			DistanceBuffer db = this.distanceBuffer;
			if(db!=null && db.target==b)return db.distance;
			this.distanceBuffer = db = new DistanceBuffer(NodeBucket.distance(this.key.key,b),b);
			return db.distance;
		}
		final public byte[] getPubKey() {
			return key.key;
		}
		final public Node getNode() {
			return node.node;
		}
		static public boolean isConnectable(String[] ss) {
			return ss!=null && ss.length>0;
		}
		@Override
		final public String[] getRemote() {
			return node.node.getRemote();
		}
		@Override
		public NormalSession getSession() {
			return node.getSession();
		}
		/////
		@Override
		public String toString() {
			if(alias==null)return Integer.toHexString(key.hashCode());
			return alias.toString();
		}
	}
	private static class DistanceBuffer{
		final int distance;
		final byte[] target;
		public DistanceBuffer(final int distance,final byte[] target) {
			this.distance = distance;
			this.target = target;
		}
	}



}
